package com.yjh;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.tools.nsc.doc.model.Public;

import java.security.PublicKey;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class mysqltomysql_allDB {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        MySqlSource<String> mySalSource = MySqlSource.<String>builder()
                .hostname("1.94.41.70")
                .port(3306)
                .databaseList("Interview")
                .tableList("Interview.*")
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> mySQL_source = env.fromSource(mySalSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
        mySQL_source.print();

        mySQL_source.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject source = jsonObject.getJSONObject("source");
                String table = source.getString("table");

                JSONObject after = jsonObject.getJSONObject("after");
                ToMysql(after,table);



                return value;
            }
        });







        env.execute();

    }

    public static void ToMysql(JSONObject after,String table)  {


        Connection connection = null;
        PreparedStatement pstmt = null;
        try {
                //        注册驱动
                Class.forName("com.mysql.cj.jdbc.Driver");
                //        获取连接
                connection = DriverManager.getConnection("jdbc:mysql://1.92.202.8:3306/Interview", "root", "000000");
                //        编写sql
                String sql = "insert into "+table+" values (?,?,?)";
            System.out.println(sql);
                //        编译sql
                pstmt = connection.prepareStatement(sql);
                //        设置参数
            System.out.println(after.getInteger("id"));
            pstmt.setInt(1,after.getInteger("id"));
                pstmt.setString(2,after.getString("problem"));
                pstmt.setString(3,after.getString("result"));
                //        运行sql
                pstmt.executeUpdate();

            } catch (Exception e) {
                System.out.println("插入失败");
            }finally {
            //        关闭连接
            try {
                pstmt.close();
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
            try {
                connection.close();
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }








    }

}
